package cc.eddic.practice.concurrency.impl;

import cc.eddic.practice.concurrency.ExecutorServiceUtil;
import cc.eddic.practice.concurrency.LogItem;
import cc.eddic.practice.concurrency.LogParser;
import lombok.extern.slf4j.Slf4j;
import lombok.val;

import java.io.*;
import java.nio.file.Files;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.zip.GZIPInputStream;

@Slf4j
public class gzipImpl implements LogParser {

    @Override
    public Map<String, List<LogItem>> parse(String folderName) {
        Map<String, List<LogItem>> map = new ConcurrentHashMap<>();
        List<File> filesList = new ArrayList<>();
        fileResult(new File(folderName), filesList);
        ExecutorService executorService = Executors.newFixedThreadPool(16);
        CountDownLatch latch = new CountDownLatch((int) filesList.stream().filter(file -> file.getName().endsWith(".gz")).count());
        filesList.stream().filter(file -> file.getName().endsWith(".gz"))
                .forEach(f -> executorService.submit(() -> ride(f, map, latch)));
        try {
            latch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage());
        }
        ExecutorServiceUtil.shutdown(executorService);
        return map;
    }


    public static void ride(File f, Map<String, List<LogItem>> map, CountDownLatch latch) {
        LogItemGZipParser logItemGZipParser = new LogItemGZipParser();
        List<String> lines = new ArrayList<>();
        try {
            InputStream in = new GZIPInputStream(Files.newInputStream(f.toPath()));
            Scanner sc = new Scanner(in);
            while (sc.hasNextLine()) {
                lines.add(sc.nextLine());
            }
            List<LogItem> logItemList = lines.stream().map(logItemGZipParser::parse)
                    .filter(Optional::isPresent).map(Optional::get).collect(Collectors.toList());
            latch.countDown();
            map.put(f.getName(), logItemList);

        } catch (IOException e) {
            log.error(e.getMessage());
        }
    }

    private static void fileResult(File file, List<File> filesList) {
        val files = file.listFiles();
        if (files == null) {
            return;
        }
        for (val f : files) {
            if (f.isDirectory()) {
                fileResult(f, filesList);
            } else filesList.add(f);
        }
    }

}
